package com.ksc.wordcount.worker;

import akka.actor.ActorRef;  // 引入Akka框架的ActorRef类，用于创建和管理Actor对象
import akka.actor.ActorSystem;  // 引入Akka框架的ActorSystem类，用于创建和管理Actor系统
import akka.actor.Props;  // 引入Akka框架的Props类，用于创建Actor的配置属性
import com.ksc.wordcount.rpc.Executor.ExecutorActor;  // 引入执行器Actor的实现类
import com.ksc.wordcount.rpc.Executor.ExecutorRpc;  // 引入执行器的RPC实现类
import com.ksc.wordcount.rpc.Executor.ExecutorSystem;  // 引入执行器的系统类
import com.ksc.wordcount.rpc.ExecutorRegister;  // 引入执行器的注册类
import com.ksc.wordcount.shuffle.nettyimpl.server.ShuffleService;  // 引入数据洗牌服务类
import com.ksc.wordcount.util.PropertyUtil;

import java.util.List;
import java.util.Map;

public class Executor {

    public static void main(String[] args) throws InterruptedException {

        //# ip         akka端口 rpc端口  内存  cpu
        //127.0.0.1   5050    7447      512m  2
        //127.0.0.1   5051    7448      512m  2
        
        // 设置执行器的环境配置信息
        // 设置执行器绑定的IP地址
//        ExecutorEnv.host = "127.0.0.1";
        ExecutorEnv.host = args[0];
        // 设置执行器绑定的端口号
//        ExecutorEnv.port = 15050;
        ExecutorEnv.port = Integer.parseInt(args[1]);

        List<Map<String, String>> masterProperties = PropertyUtil.getMasterProperties();
        Map<String, String> property = masterProperties.get(0);

        String ip = property.get("ip");
        String akkaPort = property.get("akkaPort");

        // 设置执行器连接到的驱动器地址
        ExecutorEnv.driverUrl = "akka.tcp://DriverSystem@"+ip+":"+akkaPort+"/user/driverActor";
//        ExecutorEnv.driverUrl = "akka.tcp://DriverSystem@127.0.0.1:4040/user/driverActor";

        // 设置执行器自身的地址
        ExecutorEnv.executorUrl = "akka.tcp://ExecutorSystem@" + ExecutorEnv.host + ":" + ExecutorEnv.port + "/user/executorActor";
        // 设置数据洗牌服务的端口号
//        ExecutorEnv.shufflePort = 7337;
        ExecutorEnv.shufflePort = Integer.parseInt(args[2]);

        // 设置执行器可使用的内存大小
        ExecutorEnv.memory = args[3];

        // 设置执行器的核心线程数
        ExecutorEnv.core = Integer.parseInt(args[4]);

        // 启动数据洗牌服务
        new Thread(() -> {
            try {
                new ShuffleService(ExecutorEnv.shufflePort).start();
            } catch (InterruptedException e) {
                new RuntimeException(e);
            }
        }).start();

        // 创建执行器系统
        ActorSystem executorSystem = ExecutorSystem.getExecutorSystem();
        // 创建执行器Actor
        ActorRef clientActorRef = executorSystem.actorOf(Props.create(ExecutorActor.class), "executorActor");
        // 打印执行器Actor的路径
        System.out.println("ServerActor started at: " + clientActorRef.path().toString());
        // 注册执行器到执行器注册表中
        ExecutorRpc.register(new ExecutorRegister(ExecutorEnv.executorUrl, ExecutorEnv.memory, ExecutorEnv.core));
    }


}